package com.simple.mq.core.network.core;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.Set;

/**
 * 项目: simple-mq
 * <p>
 * 功能描述: 工作线程管理器
 * 这个主要是起到阻塞的作用，分两点：
 * 1、如果当前队列有生产者，那么直接跳过take，拿到消息，准备转发给消费者，如果没有生产者，就take持续等待；这里用了java阻塞队列特性；
 * 2、如果当前队列没有消费者，这个时候就睡眠，等待消费者上线；有的话直接消费；
 *
 * @author: WuChengXing
 * @create: 2023-09-04 14:30
 **/
@Slf4j(topic = "WorkerManager")
public class WorkerManager {

    /**
     * 初始化方法，进行线程的创建并与队列绑定
     *
     * @param queueNames
     */
    public static void init(Set<String> queueNames) {
        synchronized (WorkerManager.class) {
            int queueSize = queueNames.size();
            for (String queueName : queueNames) {
                Thread thread = new Thread(new Task(queueName), "worker-" + queueName);
                // 将thread添加到每个queue的worker线程中
                QueueManager.getQueue(queueName).workers.add(thread);
                // 设置为当前线程的守护线程，随着主程序的终止而被杀死
                thread.setDaemon(true);
                thread.start();
            }
            log.info("{} worker threads are started.", queueSize);
        }
    }

    public static boolean isInited() {
        return false;
    }

    /**
     * 任务对象，持续取queue中的消息并将其转发到对应的channel中
     * 两种情况会阻塞
     * 1. queue中无元素
     * 2. queue没有consumer来绑定
     */
    private static class Task implements Runnable {

        private final Logger log = LoggerFactory.getLogger(Task.class);

        private final MessageQueue queue;

        @Override
        public void run() {
            log.debug("worker thread of queue {} is working", queue.getName());
            Object message;
            while (true) {
                try {
                    // 阻塞获取消息
                    message = queue.take();
                    List<ChannelId> channelIds;
                    while ((channelIds = BasicMaps.queueConsumerMap.get(queue.getName())) == null ||
                            channelIds.isEmpty()) {
                        // 单纯的等待消费者上线，如果一直没有消费者，这里会一直卡者
                        try {
                            // 如果没有消费者进来就先睡觉，如果消费者尝试注册，这里就唤醒，要开始工作了，需要消费消息了
                            Thread.sleep(Long.MAX_VALUE);
                            log.debug("no consumers, sleeping...");
                        } catch (InterruptedException e) {
                            //以防有人捣乱
                            log.debug("interrupted...");
                        }
                    }
                    for (ChannelId channelId : channelIds) {
                        Channel channel = BasicMaps.clients.find(channelId);
                        channel.writeAndFlush(
                                new TextWebSocketFrame(JSON.toJSONString(message))
                        ); // 发送消息
                    }
                } catch (InterruptedException e) {
                    // 不做异常处理，继续阻塞获取消息
                }
            }
        }

        // 绑定的工作队列的序号
        public Task(String queueName) {
            this.queue = QueueManager.getQueue(queueName);
        }
    }
}
